package pipeline;

import com.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException;
import model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import us.codecraft.webmagic.ResultItems;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.pipeline.Pipeline;
import util.Config;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * @author J0101
 * @Description 使用Mysql Jdbc插入数据库的不同数据库中
 */
public class MysqlPipeline implements Pipeline{
    private Logger logger = LoggerFactory.getLogger(getClass());

    private Connection conn;
    public MysqlPipeline() throws ClassNotFoundException, SQLException{
        Class.forName("com.mysql.jdbc.Driver");
        this.conn= DriverManager.getConnection(Config.getCrawler_MYSQL_URL(), Config.getCrawler_MYSQL_USER(), Config.getCrawler_MYSQL_PSWD());

    }
    @Override
    public void process(ResultItems resultItems, Task task) {
        String job_id = resultItems.getRequest().getExtra("job_id").toString();
        BaseRegisterInfo baseRegisterInfo = (BaseRegisterInfo)resultItems.get("BaseRegisterInfo");
        if(baseRegisterInfo != null){
            insertBaseRegisterInfo(baseRegisterInfo,job_id);
        }
        BaseBusinessInfo businessInfo = (BaseBusinessInfo)resultItems.get("BaseBusinessInfo");
        if(businessInfo != null){
            insertBaseBusinessInfo(businessInfo,job_id);
        }
        ArrayList<BaseMainStaff> baseMainStaffs = resultItems.get("BaseMainStaff");
        if(baseMainStaffs != null){
            insertBaseMainStaff(baseMainStaffs,job_id);
        }
        ArrayList<BaseShareholderInfo> baseShareholderInfos = resultItems.get("BaseShareholderInfo");
        if(baseShareholderInfos != null){
                insertBaseShareholderInfo(baseShareholderInfos,job_id);

        }
        ArrayList<BaseInvestment> baseInvestments = resultItems.get("BaseInvestment");
        if(baseInvestments != null){
                insertBaseInvestment(baseInvestments,job_id);

        }

        ArrayList<BaseChange> baseChanges = resultItems.get("BaseChange");
        if(baseChanges != null){
            insertBaseChange(baseChanges,job_id);

        }
        BaseProduct baseProduct = (BaseProduct) resultItems.get("BaseProduct");
        if(baseProduct != null){
            insertBaseProduct(baseProduct,job_id);
        }
        ArrayList<BaseProductList> baseProductLists = resultItems.get("BaseProductList");
        if(baseProductLists != null){
            insertBaseProductList(baseProductLists,job_id);
        }
        ArrayList<EquityPledge> equityPledges = resultItems.get("EquityPledge");
        if(equityPledges != null){
            insertEquityPledge(equityPledges,job_id);

        }
        FengxianTag fengxianTag = (FengxianTag) resultItems.get("FengxianTag");
        if(fengxianTag != null){
                insertFengxianTag(fengxianTag,job_id);

        }
        SusongTag susongTag = (SusongTag) resultItems.get("SusongTag") ;
        if(susongTag != null){
                insertSusongTag(susongTag,job_id);
        }
//        System.out.println(resultItems.get("")+"插入完成");
    }
    public void insertBaseRegisterInfo(BaseRegisterInfo baseRegisterInfo,String job_id){
        try {
            Statement state=this.conn.createStatement();
            String sql = "replace into BaseRegisterInfo(company_unique,search_word,company_name,offical_url,company_full_name,shareholder,sharehold,company_number,company_type,establishment_date,operating_status,note,wind_up_mode,fucked_up_date,register_of_charges,important_things,createdTime,job_id)" +
                    "values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象
            pstmt.setString(1,baseRegisterInfo.getCompany_unique());
            pstmt.setString(2,baseRegisterInfo.getSearch_word());
            pstmt.setString(3,baseRegisterInfo.getCompany_name());
            pstmt.setString(4,baseRegisterInfo.getOffical_url());
            pstmt.setString(5,baseRegisterInfo.getCompany_full_name());
            pstmt.setString(6,baseRegisterInfo.getShareholder());
            pstmt.setString(7,baseRegisterInfo.getSharehold());
            pstmt.setString(8,baseRegisterInfo.getCompany_number());
            pstmt.setString(9,baseRegisterInfo.getCompany_type());
            pstmt.setString(10,baseRegisterInfo.getEstablishment_date());
            pstmt.setString(11,baseRegisterInfo.getOperating_status());
            pstmt.setString(12,baseRegisterInfo.getNote());
            pstmt.setString(13,baseRegisterInfo.getWind_up_mode());
            pstmt.setString(14,baseRegisterInfo.getFucked_up_date());
            pstmt.setString(15,baseRegisterInfo.getRegister_of_charges());
            pstmt.setString(16,baseRegisterInfo.getImportant_things());
            pstmt.setString(17,baseRegisterInfo.getCreatedTime());
            pstmt.setString(18,job_id);

            pstmt.executeUpdate();//执行sql语句
            pstmt.close();
        }
        catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }

    }
    public void insertBaseBusinessInfo(BaseBusinessInfo businessInfo,String job_id){
        try {
            Statement state=this.conn.createStatement();
            String sql = "replace into BaseBusinessInfo(company_unique,search_word,company_name,credit_code,taxpayer_number,registration_number,organization_code,legal_representative_unique,legal_representative,registered_capital,operating_status,establishment_date,company_type,insured_person,person_size,operating_period,registration_authority,paid_capital,approved_date,english_name,district,industry,used_name,company_address,business_scope,offical_url,product_url,createdTime,job_id)" +
                    "values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象
            pstmt.setString(1,businessInfo.getUnique());
            pstmt.setString(2,businessInfo.getSearch_word());
            pstmt.setString(3,businessInfo.getCompany_name());
            pstmt.setString(4,businessInfo.getCredit_code());
            pstmt.setString(5,businessInfo.getTaxpayer_number());
            pstmt.setString(6,businessInfo.getRegistration_number());
            pstmt.setString(7,businessInfo.getOrganization_code());
            pstmt.setString(8,businessInfo.getLegal_representative_unique());
            pstmt.setString(9,businessInfo.getLegal_representative());
            pstmt.setString(10,businessInfo.getRegistered_capital());
            pstmt.setString(11,businessInfo.getOperating_status());
            pstmt.setString(12,businessInfo.getEstablishment_date());
            pstmt.setString(13,businessInfo.getCompany_type());
            pstmt.setString(14,businessInfo.getInsured_person());
            pstmt.setString(15,businessInfo.getPerson_size());
            pstmt.setString(16,businessInfo.getOperating_period());
            pstmt.setString(17,businessInfo.getRegistration_authority());
            pstmt.setString(18,businessInfo.getPaid_capital());
            pstmt.setString(19,businessInfo.getApproved_date());
            pstmt.setString(20,businessInfo.getEnglish_name());
            pstmt.setString(21,businessInfo.getDistrict());
            pstmt.setString(22,businessInfo.getIndustry());
            pstmt.setString(23,businessInfo.getUsed_name());
            pstmt.setString(24,businessInfo.getCompany_address());
            pstmt.setString(25,businessInfo.getBusiness_scope());
            pstmt.setString(26,businessInfo.getOffical_url());
            pstmt.setString(27,businessInfo.getProduct_url());

            pstmt.setString(28,businessInfo.getCreatedTime());
            pstmt.setString(29,job_id);
            logger.info("插入成功"+businessInfo.toString());
            pstmt.executeUpdate();//执行sql语句
            pstmt.close();
        }
        catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }

    }
    public void insertBaseProduct(BaseProduct baseProduct,String job_id){
        try {
            Statement state=this.conn.createStatement();
            String sql = "replace into BaseProduct(product_unique,product_name,product_url,financing_information," +
                    "date,address,pruduct_info,createdTime,job_id)" +
                    "values(?,?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象
            pstmt.setString(1,baseProduct.getProduct_unique());
            pstmt.setString(2,baseProduct.getProduct_name());
            pstmt.setString(3,baseProduct.getProduct_url());
            pstmt.setString(4,baseProduct.getFinancing_information());
            pstmt.setString(5,baseProduct.getDate());
            pstmt.setString(6,baseProduct.getAddress());
            pstmt.setString(7,baseProduct.getPruduct_info());
            pstmt.setString(8,baseProduct.getCreatedTime());
            pstmt.setString(9,job_id);
            pstmt.executeUpdate();//执行sql语句
            pstmt.close();
        }
        catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }

    }
    public void insertBaseProductList(ArrayList<BaseProductList> baseProductLists,String job_id ) {
        try
        {
            Statement state=this.conn.createStatement();
            String sql = "replace into BaseProductList(product_unique,main_product_name,main_product_info,main_product_link,createdTime,job_id)" +
                    "values(?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象

            for(BaseProductList baseProductList:baseProductLists){
                pstmt.setString(1,baseProductList.getProduct_unique());
                pstmt.setString(2,baseProductList.getMain_product_name());
                pstmt.setString(3,baseProductList.getMain_product_info());
                pstmt.setString(4,baseProductList.getMain_product_link());
                pstmt.setString(5,baseProductList.getCreatedTime());
                pstmt.setString(6,job_id);

                pstmt.executeUpdate();//执行sql语句
            }
            pstmt.close();
        }catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }


    }
    public void insertBaseMainStaff(ArrayList<BaseMainStaff> baseMainStaffs,String job_id ) {
        try
        {
            Statement state=this.conn.createStatement();
            String sql = "replace into BaseMainStaff(company_unique,nameUnique,name,duty,createdTime,job_id)values(?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象

            for(BaseMainStaff baseMainStaff:baseMainStaffs){
                pstmt.setString(1,baseMainStaff.getCompany_unique());
                pstmt.setString(2,baseMainStaff.getNameUnique());
                pstmt.setString(3,baseMainStaff.getName());
                pstmt.setString(4,baseMainStaff.getDuty());
                pstmt.setString(5,baseMainStaff.getCreatedTime());
                pstmt.setString(6,job_id);

                pstmt.executeUpdate();//执行sql语句
            }
            pstmt.close();
        }catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }


    }
    public void insertBaseShareholderInfo(ArrayList<BaseShareholderInfo> baseShareholderInfos,String job_id ){
        try
        {
            Statement state=this.conn.createStatement();

            String sql = "replace into BaseShareholderInfo(company_unique,shareholder_unique,shareholder,shareholding_ratio,subscription_amount,funding_date,createdTime,job_id)" +
                    "VALUES(?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象
            for(BaseShareholderInfo baseShareholderInfo:baseShareholderInfos){
                pstmt.setString(1,baseShareholderInfo.getCompany_unique());
                pstmt.setString(2,baseShareholderInfo.getShareholder_unique());
                pstmt.setString(3,baseShareholderInfo.getShareholder());
                pstmt.setString(4,baseShareholderInfo.getShareholding_ratio());
                pstmt.setString(5,baseShareholderInfo.getSubscription_amount());
                pstmt.setString(6,baseShareholderInfo.getFunding_date());
                pstmt.setString(7,baseShareholderInfo.getCreatedTime());
                pstmt.setString(8,job_id);

                pstmt.executeUpdate();//执行sql语句
            }

            pstmt.close();
        }catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }


    }
    public void insertEquityPledge(ArrayList<EquityPledge> equityPledges,String job_id ){
        try
        {
            Statement state=this.conn.createStatement();
            String sql = "replace into EquityPledge (register_id,pledgor_unique,pledgor_name,pledgee_unique,pledgee_name,stock_number,register_date,status,createdTime,job_id) values(?,?,?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象

            for(EquityPledge equityPledge:equityPledges){
                pstmt.setString(1,equityPledge.getRegister_id());
                pstmt.setString(2,equityPledge.getPledgor_unique());
                pstmt.setString(3,equityPledge.getPledgor_name());
                pstmt.setString(4,equityPledge.getPledgee_unique());
                pstmt.setString(5,equityPledge.getPledgee_name());
                pstmt.setString(6,equityPledge.getStock_number());
                pstmt.setString(7,equityPledge.getRegister_date());
                pstmt.setString(8,equityPledge.getStatus());
                pstmt.setString(9,equityPledge.getCreatedTime());
                pstmt.setString(10,job_id);

                pstmt.executeUpdate();//执行sql语句
            }

            pstmt.close();
        }catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }
    }
    public void insertBaseChange(ArrayList<BaseChange> baseChanges ,String job_id){
        try
        {
            Statement state=this.conn.createStatement();
            String sql = "replace into BaseChange(company_unique,number,change_date,change_project,before_id,before_text,before_text_em,before_text_link,after_id,after_text,after_text_em,after_text_link,createdTime,job_id)" +
                    "values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象

            for(BaseChange baseChange:baseChanges){
                pstmt.setString(1,baseChange.getCompany_unique());
                pstmt.setString(2,baseChange.getNumber());
                pstmt.setString(3,baseChange.getChange_date());
                pstmt.setString(4,baseChange.getChange_project());
                pstmt.setString(5,baseChange.getBefore_id());
                pstmt.setString(6,baseChange.getBefore_text());
                pstmt.setString(7,baseChange.getBefore_text_em());
                pstmt.setString(8,baseChange.getBefore_text_link());
                pstmt.setString(9,baseChange.getAfter_id());
                pstmt.setString(10,baseChange.getAfter_text());
                pstmt.setString(11,baseChange.getAfter_text_em());
                pstmt.setString(12,baseChange.getAfter_text_link());
                pstmt.setString(13,baseChange.getCreatedTime());
                pstmt.setString(14,job_id);

                pstmt.executeUpdate();//执行sql语句
            }

            pstmt.close();
        }catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }


    }
    public void insertBaseInvestment(ArrayList<BaseInvestment> baseInvestments ,String job_id){
        try
        {
            Statement state=this.conn.createStatement();
            String sql = "replace into BaseInvestment(company_unique,investment_unique,investment_name,investment_people,registered_capital,funded_ratio,establishment_date,status,createdTime,job_id)" +
                    "values(?,?,?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象

            for(BaseInvestment baseInvestment:baseInvestments){
                pstmt.setString(1,baseInvestment.getCompany_unique());
                pstmt.setString(2,baseInvestment.getInvestment_unique());
                pstmt.setString(3,baseInvestment.getInvestment_name());
                pstmt.setString(4,baseInvestment.getInvestment_people());
                pstmt.setString(5,baseInvestment.getRegistered_capital());
                pstmt.setString(6,baseInvestment.getFunded_ratio());
                pstmt.setString(7,baseInvestment.getEstablishment_date());
                pstmt.setString(8,baseInvestment.getStatus());
                pstmt.setString(9,baseInvestment.getCreatedTime());
                pstmt.setString(10,job_id);

                pstmt.executeUpdate();//执行sql语句
            }

            pstmt.close();
        }catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }


    }
    public void insertFengxianTag(FengxianTag fengxianTag ,String job_id){
        try
        {
            Statement state=this.conn.createStatement();
            String sql = "replace into FengxianTag(company_unique,jingyingyichang,yangzhongweifa,guquanchuzhi,xingzhengchufa,shuishouweifa,dongchandiya,qingsuanxinxi,sifapaimai,tudizhiya,jianyizhuxiao,gongshicuigao,qianshuigonggao,createdTime,job_id) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象
            pstmt.setString(1,fengxianTag.getCompany_unique());
            pstmt.setString(2,fengxianTag.getJingyingyichang());
            pstmt.setString(3,fengxianTag.getYangzhongweifa());
            pstmt.setString(4,fengxianTag.getGuquanchuzhi());
            pstmt.setString(5,fengxianTag.getXingzhengchufa());
            pstmt.setString(6,fengxianTag.getShuishouweifa());
            pstmt.setString(7,fengxianTag.getDongchandiya());
            pstmt.setString(8,fengxianTag.getQingsuanxinxi());
            pstmt.setString(9,fengxianTag.getSifapaimai());
            pstmt.setString(10,fengxianTag.getTudizhiya());
            pstmt.setString(11,fengxianTag.getJianyizhuxiao());
            pstmt.setString(12,fengxianTag.getGongshicuigao());
            pstmt.setString(13,fengxianTag.getQianshuigonggao());
            pstmt.setString(14,fengxianTag.getCreatedTime());
            pstmt.setString(15,job_id);

            pstmt.executeUpdate();//执行sql语句
            pstmt.close();
        }catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }

    }
    public void insertSusongTag(SusongTag susongTag ,String job_id) {
        try
        {
            Statement state=this.conn.createStatement();
            String sql = "replace into SusongTag(company_unique,zhixing,shixin,wenshu,fayuan,kaiting,songda,sifa,createdTime,job_id)values(?,?,?,?,?,?,?,?,?,?)";
            PreparedStatement pstmt=this.conn.prepareStatement(sql);//获得预置对象
            pstmt.setString(1,susongTag.getCompany_unique());
            pstmt.setString(2,susongTag.getZhixing());
            pstmt.setString(3,susongTag.getShixin());
            pstmt.setString(4,susongTag.getWenshu());
            pstmt.setString(5,susongTag.getFayuan());
            pstmt.setString(6,susongTag.getKaiting());
            pstmt.setString(7,susongTag.getSongda());
            pstmt.setString(8,susongTag.getSifa());
            pstmt.setString(9,susongTag.getCreatedTime());
            pstmt.setString(10,job_id);

            pstmt.executeUpdate();//执行sql语句
            pstmt.close();
        }catch ( MySQLIntegrityConstraintViolationException ee){
            logger.warn("已有记录");
        }catch (SQLException e){
            System.out.println(e);
        }

    }

}
